//
// Created by dave on 18-8-5.
//
#include <atomic>
#include <vector>
#include <thread>
#include <random>
#include <glog/logging.h>
#include <evpp/buffer.h>
#include <evpp/tcp_conn.h>
#include <evpp/tcp_client.h>
#include <evpp/event_loop_thread_pool.h>
#include "codec/data_pack.h"
#include "codec/data_utils.h"
#include "codec/terminal/term_register.h"
#include "codec/terminal/location_report.h"

size_t _find_mark(const char* ptr, size_t start, size_t stop) {
    for (size_t i = start; i < stop; i++) {
        if (ptr[i] == 0x7E) {
            return i;
        }
    }
    return 0;
}

class ClientGroup {
public:
    ClientGroup(evpp::EventLoop* loop,
                const std::string& serverAddr, /*ip:port*/
                int connectCount,
                int threadCount)
            : connectCount_(connectCount), loop_(loop), tpool_(std::make_shared<evpp::EventLoopThreadPool>(loop, threadCount)) {
        tpool_->Start(true);

        for (int i = 0; i < connectCount; i++) {
            auto c = std::make_shared<evpp::TCPClient>(tpool_->GetNextLoop(), serverAddr, "client" + std::to_string(i + 1));
            c->SetConnectionCallback(std::bind(&ClientGroup::onConnect, this, std::placeholders::_1));
            c->SetMessageCallback(std::bind(&ClientGroup::onMessage, this, std::placeholders::_1, std::placeholders::_2));
            c->set_connecting_timeout(evpp::Duration(10.0));
            c->Connect();
            clients_.push_back(c);
        }
    }

    std::vector<std::shared_ptr<evpp::TCPClient>>& getClients() {
        return clients_;
    }

    int getConnected() {
        return connected_.load();
    }

private:
    void onConnect(const evpp::TCPConnPtr& conn) {
        if (conn->IsConnected()) {
            int c = ++connected_;
            conn->SetTCPNoDelay(true);
            LOG(INFO) << "已连接：" << c;
        } else {
            --connected_;
        }
    }

    void onMessage(const evpp::TCPConnPtr& conn, evpp::Buffer* msg) {
        size_t size = msg->length();
        if (size > 4 * 1024) {
            msg->Reset();
            return;
        }

        const char* p = msg->data();
        size_t start = _find_mark(p, 0, size);

        if (start == 0 && p[0] != 0x7E) {
            msg->Reset();
        } else {
            do {
                size_t end = _find_mark(p, start + 1, size);
                if (end > 0) {
                    if (end - start == 1) {
                        start = end;
                        continue;
                    }

                    onFrame(conn, p + start, end - start + 1);
                    start = _find_mark(p, end + 1, size);
                    if (start == 0) {
                        msg->Skip(end + 1);
                        break;
                    }
                } else {
                    msg->Skip(start);
                    break;
                }
            } while (true);
        }
    }

    void onFrame(const evpp::TCPConnPtr &conn, const char *ptr, size_t length) {
        int64_t count = ++msgCounter_;
        if (count % 20000 == 0) {
            LOG(INFO) << "接收(万包)：" << count / 10000;
        }

        if (length < 14) {
            LOG(WARNING) << "无效桢：" << length << "字节";
            return;
        }

        DataPack dp = decode_data_pack(ptr, 0, length);
        // LOG(INFO) << "接收：" << std::hex << dp.getMsgId() << ":" << std::dec << dp.getSeqNo() << ":" << dp.getSim();
    }

private:
    int connectCount_;
    std::atomic_int connected_{0};
    std::atomic_int64_t msgCounter_{0};
    evpp::EventLoop* loop_;
    std::shared_ptr<evpp::EventLoopThreadPool> tpool_;
    std::vector<std::shared_ptr<evpp::TCPClient>> clients_;
};

int main(int argc, char** argv) {
    google::InitGoogleLogging(argv[0]);
    FLAGS_stderrthreshold = 0;

    int connectCount = 6000;
    evpp::EventLoop loop;
    ClientGroup group(&loop, "192.168.3.4:2048", connectCount, 8);
    std::vector<std::thread> vec;

    int threadCount = 6;
    std::atomic_int64_t sendCount(0);
    std::random_device rand;
    std::default_random_engine engine(rand());
    std::uniform_int_distribution<> dis(0, 10000);
    auto dice = std::bind(dis, engine);
    for (int i = 0; i < threadCount; i++) {
        vec.emplace_back(std::thread([&, g = i] {
            if (group.getConnected() < connectCount) {
                std::this_thread::sleep_for(std::chrono::seconds(1));
            }

            LOG(INFO) << "3秒钟后开始发送数据";
            std::this_thread::sleep_for(std::chrono::seconds(3));
            auto startTime = std::chrono::system_clock::now();
            int count = connectCount / threadCount;
            unsigned short seq = 1;

#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wmissing-noreturn"
            while (true) {
                for (int j = count * g; j < count * (g + 1); j++) {
                    evpp::TCPConnPtr conn = group.getClients()[j]->conn();
                    std::string sim = string_format("0188%08d", j + 1);
                    DataPack re;
                    re.setMsgId(0x0100); // 注册
                    re.setSeqNo(seq);
                    re.setSim(sim);

                    auto body = std::make_shared<TermRegister>();
                    body->setProvince(42);
                    body->setCity(100);
                    body->setMaker("abcde");
                    body->setModel("12345678");
                    body->setDeviceId("abcdefg");
                    body->setColor(2);
                    body->setLicense("鄂A12345");
                    re.setBody(body);
                    std::vector<char> bytes = re.encode();
                    conn->Send(bytes.data(), bytes.size());

                    re.setMsgId(0x0200); // 上报位置
                    re.setSeqNo(static_cast<unsigned short>(seq+ 1));
                    auto loc = std::make_shared<LocationReport>();
                    loc->setAlarm(0);
                    loc->setFlag(2);
                    loc->setLat(30 + dice() / 10000.0);
                    loc->setLng(118 + dice() / 10000.0);
                    loc->setHeight(static_cast<unsigned short>(0.15 * dice()));
                    loc->setSpeed(static_cast<unsigned short>(0.08 * dice()));
                    loc->setDirection(static_cast<unsigned short>(0.036 * dice()));
                    loc->setTime(std::time(nullptr));
                    re.setBody(loc);
                    bytes = re.encode();
                    conn->Send(bytes.data(), bytes.size());

                    long c = sendCount += 2;
                    if (c % 20000 == 0) {
                        auto end = std::chrono::system_clock::now();
                        auto diff = std::chrono::duration_cast<std::chrono::milliseconds>(end - startTime);
                        LOG(INFO) << "已发出（万包）：" << c / 10000 << ", " << diff.count();
                    }
                }

                seq += 2;
                if (++seq >= 0xFFFF - 1) {
                    seq = 1;
                }

                // std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
#pragma clang diagnostic pop
#pragma clang diagnostic pop

        }));
    }


    loop.Run();
    return 0;
}